package com.cet.kafkaconsumer.listener;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * @program: cet-practice
 * @description:
 * @author: 陈恩涛
 * @create: 2023-07-24 10:39
 **/

@Slf4j
@Component
public class KafkaConsumerListener {

//    @KafkaListener(topics = "topic1")
//    public void onMessageSimple(ConsumerRecord<?, ?> record) {
//        // 消费的哪个topic、partition的消息,打印出消息内容
//        System.out.println("简单消费：" + record.topic() + "-" + record.partition() + "-" + record.value());
//    }

    @KafkaListener(topicPartitions = {
        @TopicPartition(topic = "first", partitions = {"0"})
    })
    public void onMessage(ConsumerRecord<?, ?> consumerRecord) {
        Optional<?> optional = Optional.ofNullable(consumerRecord.value());
        if (optional.isPresent()) {
            Object msg = optional.get();
            log.info("partition=0,message:[{}]", msg);
        }
    }

    @KafkaListener(topicPartitions = {
        @TopicPartition(topic = "first", partitions = {"1"})
    })
    public void onMessage1(ConsumerRecord<?, ?> consumerRecord) {
        Optional<?> optional = Optional.ofNullable(consumerRecord.value());
        if (optional.isPresent()) {
            Object msg = optional.get();
            log.info("partition=1,message:[{}]", msg);
        }
    }

    @KafkaListener(topicPartitions = {
        @TopicPartition(topic = "first", partitions = {"2"})
    })
    public void onMessage2(ConsumerRecord<?, ?> consumerRecord) {
        Optional<?> optional = Optional.ofNullable(consumerRecord.value());
        if (optional.isPresent()) {
            Object msg = optional.get();
            log.info("partition=2,message:[{}]", msg);
        }
    }
}
